package redis.clients.jedis;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.slf4j.LoggerFactory;

import redis.clients.jedis.exceptions.JedisConnectionException;
import redis.clients.util.Pool;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
/**
 * 目前只是提供了思路，需要大量的优化这个类，怎么去维护多个从的连接等等，
 * @author Han
 */
public class JedisSentinelSlavePool extends Pool<Jedis> {
	private static final org.slf4j.Logger logger = LoggerFactory.getLogger(JedisSentinelSlavePool.class);
	private volatile HostAndPort currentSlave;
	protected GenericObjectPoolConfig poolConfig;
	
	private Set<String> sentinels;
	private String masterName;
	
	protected int connectionTimeout = Protocol.DEFAULT_TIMEOUT;
	protected int soTimeout = Protocol.DEFAULT_TIMEOUT;

	protected String password;

	protected int database = Protocol.DEFAULT_DATABASE;

	protected String clientName;

	protected Set<SlaveListener> masterListeners = new HashSet<SlaveListener>();

	protected Logger log = Logger.getLogger(getClass().getName());

	private volatile JedisFactory factory;
	private volatile HostAndPort currentHostMaster;

	public JedisSentinelSlavePool(String masterName, Set<String> sentinels, final GenericObjectPoolConfig poolConfig) {
		this(masterName, sentinels, poolConfig, Protocol.DEFAULT_TIMEOUT, null, Protocol.DEFAULT_DATABASE);
	}

	public JedisSentinelSlavePool(String masterName, Set<String> sentinels) {
		this(masterName, sentinels, new GenericObjectPoolConfig(), Protocol.DEFAULT_TIMEOUT, null, Protocol.DEFAULT_DATABASE);
	}

	public JedisSentinelSlavePool(String masterName, Set<String> sentinels, String password) {
		this(masterName, sentinels, new GenericObjectPoolConfig(), Protocol.DEFAULT_TIMEOUT, password);
	}

	public JedisSentinelSlavePool(String masterName, Set<String> sentinels, final GenericObjectPoolConfig poolConfig, int timeout, final String password) {
		this(masterName, sentinels, poolConfig, timeout, password, Protocol.DEFAULT_DATABASE);
	}

	public JedisSentinelSlavePool(String masterName, Set<String> sentinels, final GenericObjectPoolConfig poolConfig, final int timeout) {
		this(masterName, sentinels, poolConfig, timeout, null, Protocol.DEFAULT_DATABASE);
	}

	public JedisSentinelSlavePool(String masterName, Set<String> sentinels, final GenericObjectPoolConfig poolConfig, final String password) {
		this(masterName, sentinels, poolConfig, Protocol.DEFAULT_TIMEOUT, password);
	}

	public JedisSentinelSlavePool(String masterName, Set<String> sentinels, final GenericObjectPoolConfig poolConfig, int timeout, final String password, final int database) {
		this(masterName, sentinels, poolConfig, timeout, timeout, password, database);
	}

	public JedisSentinelSlavePool(String masterName, Set<String> sentinels, final GenericObjectPoolConfig poolConfig, int timeout, final String password, final int database, final String clientName) {
		this(masterName, sentinels, poolConfig, timeout, timeout, password, database, clientName);
	}

	public JedisSentinelSlavePool(String masterName, Set<String> sentinels, final GenericObjectPoolConfig poolConfig, final int timeout, final int soTimeout, final String password, final int database) {
		this(masterName, sentinels, poolConfig, timeout, soTimeout, password, database, null);
	}

	public JedisSentinelSlavePool(String masterName, Set<String> sentinels, final GenericObjectPoolConfig poolConfig, final int connectionTimeout, final int soTimeout, final String password,
			final int database, final String clientName) {
		this.poolConfig = poolConfig;
		this.connectionTimeout = connectionTimeout;
		this.soTimeout = soTimeout;
		this.password = password;
		this.database = database;
		this.clientName = clientName;
		this.sentinels = sentinels;
		this.masterName = masterName;
		initSlavePool();
	}
	
	public void initSlavePool(){
		List<HostAndPort> slaves = initSentinels(sentinels, masterName);
		initPool(slaves.get(0));
	}
	
	public void destroy() {
		for (SlaveListener m : masterListeners) {
			m.shutdown();
		}

		super.destroy();
	}

	public HostAndPort getCurrentHostMaster() {
		return currentHostMaster;
	}

	private void initPool(HostAndPort slave) {
		//先测试只用第一个
		if(!slave.equals(currentSlave)){
			currentSlave = slave;
			System.out.println("执行初始化....");
			if (factory == null) {
				factory = new JedisFactory(slave.getHost(), slave.getPort(), connectionTimeout, soTimeout, password, database, clientName);
				initPool(poolConfig, factory);
			} else {
				factory.setHostAndPort(currentHostMaster);
				// although we clear the pool, we still have to check the
				// returned object
				// in getResource, this call only clears idle instances, not
				// borrowed instances
				internalPool.clear();
			}
		}
		
	}


	private List<HostAndPort> initSentinels(Set<String> sentinels, final String masterName) {
		List<HostAndPort> result = new ArrayList<HostAndPort>();
		log.info("Trying to find master from available Sentinels...");
		for (String sentinel : sentinels) {
			final HostAndPort hap = toHostAndPort(Arrays.asList(sentinel.split(":")));
			log.fine("Connecting to Sentinel " + hap);

			Jedis jedis = null;
			try {
				jedis = new Jedis(hap.getHost(), hap.getPort());

				List<Map<String, String>> slaves = jedis.sentinelSlaves(masterName);
				// debug信息
				if (logger.isDebugEnabled()) {
					try {
						System.out.println(new ObjectMapper().writeValueAsString(slaves));
					} catch (JsonProcessingException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
				}
				if (slaves == null) {
					log.warning("Can not get slaves addr, master name: " + masterName + ". Sentinel: " + hap + ".");
					continue;
				}

				for (Map<String, String> slave : slaves) {
					// 从当前可用
					if ("slave".equals(slave.get("flags")) && "ok".equals(slave.get("master-link-status")) && StringUtils.isNumeric(slave.get("port"))) {
						HostAndPort node = new HostAndPort(slave.get("ip"), Integer.parseInt(slave.get("port")));
						result.add(node);
					}
				}

				break;
			} catch (JedisConnectionException e) {
				log.warning("Cannot connect to sentinel running @ " + hap + ". Trying next one.");
			} finally {
				if (jedis != null) {
					jedis.close();
				}
			}
		}

		for (String sentinel : sentinels) {
			final HostAndPort hap = toHostAndPort(Arrays.asList(sentinel.split(":")));
			SlaveListener masterListener = new SlaveListener(masterName, hap.getHost(), hap.getPort());
			// whether MasterListener threads are alive or not, process can be stopped
			masterListener.setPool(this);
			masterListener.setDaemon(true);
			masterListeners.add(masterListener);
			masterListener.start();
		}
		
		return result;
	}

	private HostAndPort toHostAndPort(List<String> getMasterAddrByNameResult) {
		String host = getMasterAddrByNameResult.get(0);
		int port = Integer.parseInt(getMasterAddrByNameResult.get(1));

		return new HostAndPort(host, port);
	}

	@Override
	public Jedis getResource() {
		while (true) {
			Jedis jedis = super.getResource();
			jedis.setDataSource(this);

			// get a reference because it can change concurrently
			final HostAndPort master = currentHostMaster;
			final HostAndPort connection = new HostAndPort(jedis.getClient().getHost(), jedis.getClient().getPort());

			if (master.equals(connection)) {
				// connected to the correct master
				return jedis;
			} else {
				jedis.close();
			}
		}
	}

	/**
	 * @deprecated starting from Jedis 3.0 this method won't exist. Resouce cleanup should be done
	 *             using @see {@link redis.clients.jedis.Jedis#close()}
	 */
	public void returnBrokenResource(final Jedis resource) {
		if (resource != null) {
			returnBrokenResourceObject(resource);
		}
	}

	/**
	 * @deprecated starting from Jedis 3.0 this method won't exist. Resouce cleanup should be done
	 *             using @see {@link redis.clients.jedis.Jedis#close()}
	 */
	public void returnResource(final Jedis resource) {
		if (resource != null) {
			resource.resetState();
			returnResourceObject(resource);
		}
	}

	protected class SlaveListener extends Thread {

		protected String masterName;
		protected String host;
		protected int port;
		protected long subscribeRetryWaitTimeMillis = 5000;
		protected volatile Jedis j;
		protected AtomicBoolean running = new AtomicBoolean(false);
		protected JedisSentinelSlavePool pool;
		
		public JedisSentinelSlavePool getPool() {
			return pool;
		}

		public void setPool(JedisSentinelSlavePool pool) {
			this.pool = pool;
		}

		protected SlaveListener() {
		}

		public SlaveListener(String masterName, String host, int port) {
			super(String.format("MasterListener-%s-[%s:%d]", masterName, host, port));
			this.masterName = masterName;
			this.host = host;
			this.port = port;
		}

		public SlaveListener(String masterName, String host, int port, long subscribeRetryWaitTimeMillis) {
			this(masterName, host, port);
			this.subscribeRetryWaitTimeMillis = subscribeRetryWaitTimeMillis;
		}

		public void run() {

			running.set(true);

			while (running.get()) {

				j = new Jedis(host, port);

				try {
					// double check that it is not being shutdown
					if (!running.get()) {
						break;
					}
					//订阅slave主观上线下线
					j.subscribe(new JedisPubSub() {
						@Override
						public void onMessage(String channel, String message) {
							logger.info("Sentinel " + host + ":" + port + " published: " + message + ".");

							String[] switchMasterMsg = message.split(" ");

							if (switchMasterMsg.length > 3 && message.contains("@")) {
								//应该有更细致的判断
								pool.initSlavePool();
							} else {
								log.severe("Invalid message received on Sentinel " + host + ":" + port + " on channel +switch-master: " + message);
							}
						}
					} ,"+sdown","-sdown");

				} catch (JedisConnectionException e) {

					if (running.get()) {
						log.severe("Lost connection to Sentinel at " + host + ":" + port + ". Sleeping 5000ms and retrying.");
						try {
							Thread.sleep(subscribeRetryWaitTimeMillis);
						} catch (InterruptedException e1) {
							e1.printStackTrace();
						}
					} else {
						log.fine("Unsubscribing from Sentinel at " + host + ":" + port);
					}
				} finally {
					j.close();
				}
			}
		}

		public void shutdown() {
			try {
				log.fine("Shutting down listener on " + host + ":" + port);
				running.set(false);
				// This isn't good, the Jedis object is not thread safe
				if (j != null) {
					j.disconnect();
				}
			} catch (Exception e) {
				log.log(Level.SEVERE, "Caught exception while shutting down: ", e);
			}
		}
	}
}